package eip.chapter9.splitaggregate.impl;

import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mule.impl.MuleEvent;
import org.mule.impl.MuleMessage;
import org.mule.routing.AggregationException;
import org.mule.routing.inbound.CorrelationAggregator;
import org.mule.routing.inbound.EventGroup;
import org.mule.umo.UMOMessage;
import org.mule.umo.transformer.TransformerException;

import eip.chapter9.splitaggregate.AggregatedResponse;
import eip.chapter9.splitaggregate.HotelInfo;
import eip.chapter9.splitaggregate.ReservationInfo;
import eip.chapter9.splitaggregate.UserInfo;

public class InformationAggregator extends CorrelationAggregator {
	
	private static final Log LOG = LogFactory.getLog(InformationAggregator.class); 

	/**
	 * This receives three serialized events (java classes). We put them together
	 * here to create one outgoing message.
	 */
	@Override
	protected UMOMessage aggregateEvents(EventGroup events) throws AggregationException {
		LOG.info("All messages received to aggregate:" + events);
		
		AggregatedResponse response = new AggregatedResponse();
		response.setId(events.getGroupId().toString());
		
		for (Iterator iter = events.iterator(); iter.hasNext();) {
			MuleEvent event = (MuleEvent) iter.next();
			Object payload = null;
			try {
				payload = event.getTransformedMessage();
				if (payload instanceof HotelInfo) {
					response.setHotelInfo((HotelInfo) payload);
				} else if (payload instanceof ReservationInfo) {
					response.setReservationInfo((ReservationInfo) payload);
				} else if (payload instanceof UserInfo) {
					response.setUserInfo((UserInfo) payload);	
				} else {
					LOG.warn("Unparsable response received:" + payload);		
				}				
			} catch (TransformerException e) {
				LOG.warn("Unparsable response received:" + payload);
			}
		}

		return new MuleMessage(response);
	}
}
